package com.rrd.hive.udf;

import com.weicai.caesar.CaesarUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;


@Description(
        name = "mobile_data",
        value = "_FUNC_(mobile) - Get area string from mobile or caesar encode mobile",
        extended = "Example:\n> SELECT _FUNC_('13811360000');\n  {\"city_county_id\":\"110000\",\"lng\":\"116.407526\",\"zip_code\":\"100000\",\"isp\":\"中国移动\",\"prefix\":\"138\",\"provice_id\":\"110000\",\"asc_isp\":\"\",\"city_county_simple\":\"北京\",\"provice_simple\":\"北京\",\"paragraph\":\"1381136\",\"provice\":\"北京市\",\"city_county\":\"北京市\",\"zone_description\":\"010\",\"lat\":\"39.90403\"} \n> SELECT _FUNC_('13811360000')['city_county_simple'];\n北京"
)
public class MobileDataUDF extends UDF {
    public static final Map<String, Map<String, String>> areaMap = new HashMap<>();
    public static String dataFile = "/data/dim/data_phone/data_phone.csv";

    static {
        try {
            readAreaFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static Configuration config() {
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        return conf;
    }

    public static void readAreaFile() throws IOException {
        Configuration conf = config();
        FileSystem fs = FileSystem.newInstance(URI.create(dataFile), conf);
        Path remotePath = new Path(dataFile);
        FSDataInputStream in = fs.open(remotePath);
        if (fs.exists(remotePath)) {
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
            String lineTxt;

            String key = bufferedReader.readLine();
            List<String> keyList = Arrays.stream(key.replace("\"", "").split(",")).collect(Collectors.toList());


            while ((lineTxt = bufferedReader.readLine()) != null) {
                List<String> valueList = Arrays.stream(lineTxt.replace("\"", "").split(",")).collect(Collectors.toList());
                Map<String, String> lineMap = IntStream
                        .range(0, keyList.size())
                        .collect(HashMap::new,
                                (m, i) -> m.put(keyList.get(i), i >= valueList.size() ? "" : valueList.get(i)),
                                (m1, m2) -> {
                                });

                areaMap.put(lineMap.get("paragraph"), lineMap);
            }
            bufferedReader.close();
            in.close();
            fs.close();
        } else {
            System.out.println("找不到指定的文件");
        }
    }

    public Map<String, String> evaluate(String mobile) {
        if (mobile == null || mobile.equals("")) {
            return null;
        }
        if (CaesarUtil.isEncrypted(mobile)) {
            mobile = CaesarUtil.decode(mobile);
        }
        if (mobile.length() != 11) {
            return null;
        }
        return areaMap.get(mobile.substring(0, 7));
    }

}
